
package org.sopt.dev.ImgKnight.frontier.Controller;

import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;
import org.apache.commons.collections.set.SynchronizedSet;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

import org.sopt.dev.ImgKnight.frontier.Frontier;
import org.sopt.dev.ImgKnight.frontier.Executor.UrlPicker;
import org.sopt.dev.ImgKnight.frontier.NIO.NioObjectStreamHandler;
import org.sopt.dev.ImgKnight.frontier.Queue.SiteMapHashTable;

public class AgentController {
	
	private Frontier frontier = null;
	
	private SiteMapHashTable siteMapHashTable = null;
	
	private Channel listenForAgentChannel = null;
	
	private Set<Channel> agentChannelSet = null;
	
	private Set<Channel> agentReadyChannelSet = null;
	
	private ExecutorService nioBossExecutor = null;
	
	private ExecutorService nioWorkerExecutor = null;
	
	private Thread urlPicker = null;
	
	private Logger serviceLogger = null;
	
	private Logger exceptionLogger = null;
	
	@SuppressWarnings("unchecked")
	public AgentController(SiteMapHashTable siteMapHashTable) {
		frontier = Frontier.getInstance();
		agentChannelSet = SynchronizedSet.decorate(new HashSet<Channel>(frontier.getFrontierConfig().MaxCrawlerConnectionSize));
		agentReadyChannelSet = SynchronizedSet.decorate(new HashSet<Channel>(frontier.getFrontierConfig().MaxCrawlerConnectionSize));
		this.siteMapHashTable = siteMapHashTable;
		urlPicker = Executors.defaultThreadFactory()
			.newThread(new UrlPicker(this, frontier.getFrontierConfig().UrlPickerDelayMilliSec));
		
		serviceLogger = Logger.getLogger("frontier.service");
		exceptionLogger = Logger.getLogger("frontier.exception");
	}
	
	public void open() throws IllegalArgumentException {
		nioBossExecutor = Executors.newCachedThreadPool();
		nioWorkerExecutor = Executors.newCachedThreadPool();
		ChannelFactory factory 
			= new NioServerSocketChannelFactory(nioBossExecutor, nioWorkerExecutor);
		ServerBootstrap bootstrap = new ServerBootstrap(factory);
		NioObjectStreamHandler handler = new NioObjectStreamHandler();
		bootstrap.getPipeline().addLast("handler", handler);
		bootstrap.setOption("child.tcpNoDelay", true);
		bootstrap.setOption("child.keepAlive", true);
		listenForAgentChannel = bootstrap.bind(new InetSocketAddress(frontier.getFrontierConfig().ServicePort));
		if( !listenForAgentChannel.isBound() )
			throw new IllegalArgumentException("Service port: "+frontier.getFrontierConfig().ServicePort+" Bound Exception");
		urlPicker.start();
	}
	
	public void close() {
		agentChannelSet.clear();
		agentReadyChannelSet.clear();
		if( nioBossExecutor != null )
			nioBossExecutor.shutdownNow();
		if( nioWorkerExecutor != null )
			nioWorkerExecutor.shutdownNow();
		if( listenForAgentChannel != null )
			listenForAgentChannel.close().awaitUninterruptibly();
		if( urlPicker != null )
			urlPicker.interrupt();
	}
	
	public void onConnectionEstablished(Channel channel) {
		if( agentChannelSet.size() < frontier.getFrontierConfig().MaxCrawlerConnectionSize ) {
			serviceLogger.info("Agent accepted from: "+channel.getRemoteAddress().toString());
			agentChannelSet.add(channel);
		}
		else {
			exceptionLogger.warn("Connection is full. This connection will be closed:"
					+channel.getRemoteAddress().toString());
			channel.close();
		}
	}
	
	public void onConnectionClosed(Channel channel) {
		serviceLogger.info("Agent disconnected from: "+channel.getRemoteAddress().toString());
		channel.close();
		agentChannelSet.remove(channel);
		agentReadyChannelSet.remove(channel);
	}
	
	@SuppressWarnings("unchecked")
	public void onRecvedObject(Channel channel, Object obj) {
		if( obj instanceof List ) {
			List<String> extracted = (List<String>)obj;
			int pushedCount = siteMapHashTable.offerToCache(extracted);
			serviceLogger.debug("Push: "+pushedCount+" links (Total: "+siteMapHashTable.getTotalPushedCount()+")");
		} 
		else if( obj instanceof Integer ) {
			Integer r = (Integer)obj;
			if( r.intValue() == Frontier.REQJOIN )
				agentReadyChannelSet.add(channel);
			else if( r.intValue() == Frontier.REQDROP )
				agentReadyChannelSet.remove(channel);
			else 
				exceptionLogger.warn("Unexpected signum: " + obj.toString());
		}
		else {
			exceptionLogger.warn("Unexpected object: " + obj.toString());
			channel.close();
		}
	}
	
	public Channel[] getCurrentChannel() {
		Channel[] c = new Channel[agentChannelSet.size()];
		agentChannelSet.toArray(c);
		return c;
	}
	
	@Deprecated
	public void BroadCastMessage(int signum) {
		Iterator<Channel> it = agentChannelSet.iterator();
		while( it.hasNext() )
			it.next().write(new Integer(signum));
	}
	
	public SiteMapHashTable getSiteMapHashTable() {
		return siteMapHashTable;
	}
	
	public Channel[] getCurrentReadyChannel() {
		Channel[] c = null;
		c = new Channel[agentReadyChannelSet.size()];
		agentReadyChannelSet.toArray(c);
		return c;
	}
	
	public void FillAgentReadyChannelSet() {
		Iterator<Channel> it = agentChannelSet.iterator();
		while( it.hasNext() )
			agentReadyChannelSet.add(it.next());
		serviceLogger.info("Fill agentReadyChannelSet: "+agentReadyChannelSet.size());
	}
	
	public void closeAgentReadyChannelSet() {
		serviceLogger.info("Clear agentReadyChannelSet: "+agentReadyChannelSet.size());
		agentReadyChannelSet.clear();
	}
	
	public long getTransferredMessages() {
		return NioObjectStreamHandler.getTransferredMessages();
	}
	
}
